import bean.Order;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class Demo01 {

    public static void main(String[] args) throws Exception {

        // 1 env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // 2 source
        DataStreamSource<Object> orderDS = env.addSource(new RichSourceFunction<Object>() {

            private Boolean isRunning = true;

            @Override
            public void run(SourceContext<Object> sourceContext) throws Exception {
                Random random = new Random();
                while (isRunning) {
                    Order order = new Order(UUID.randomUUID().toString(), random.nextInt(3), random.nextInt(101), System.currentTimeMillis());
                    TimeUnit.SECONDS.sleep(1);
                    sourceContext.collect(order);
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        // 3 transformation
//        orderDS.assignTimestampsAndWatermarks(WatermarkStrategy
//                .forBoundedOutOfOrderness(Duration.ofSeconds(2))
//                .withTimestampAssigner())

        orderDS.print();
        env.execute();

    }


}
